package com.shujia.wyh.base;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.regionserver.BloomType;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

/*
    使用java api的方式操作Hbase数据库的基本用法

    1、本质上是JDBC
    2、获取与HBase数据库的连接对象
    3、获取数据库表操作对象
    4、增删改查
    5、释放资源

    操作：
        1、创建表
        2、删除表
        3、添加数据
            添加一条数据
            添加一批数据
        4、查询
            查询某一条数据
            查询多条数据
            根据指定的条件查询数据
        5、预分region表

 */
public class HbaseClientDemo {
    private Connection conn;
    private Admin admin;

    @Before
    public void init() {
        try {
            //获取与Hbase的连接对象
            //1、获取Hadoop的相关环境
            Configuration conf = new Configuration();

            //2、设置以下zookeeper的集群信息
            conf.set("hbase.zookeeper.quorum", "master:2181,node1:2181,node2:2181");

            //3、获取habse的连接对象
            conn = ConnectionFactory.createConnection(conf);

            //4、获取HMaster的对象
            admin = conn.getAdmin();

            System.out.println("获取连接成功！！" + conn);
            System.out.println("HMaster的对象获取成功！！" + admin);
            System.out.println("------------------------------------------------------------------");


        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 需求1：在hbase中创建一张表
     */
    @Test
    public void createTable() {
        try {
            //在Hbase中创建一张表必须项是表名和列簇名
            //表在API中对应一个表描述器单独创建，列簇也对应一个列簇描述器，单独创建
//            HTableDescriptor
            TableDescriptorBuilder test2 = TableDescriptorBuilder.newBuilder(TableName.valueOf("students"));
//            HColumnDescriptor
            ColumnFamilyDescriptor info = ColumnFamilyDescriptorBuilder.of("info");
            //创建列簇的同时去设置布隆过滤器
//            ColumnFamilyDescriptor info = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes("info"))
//                    .setBloomFilterType(BloomType.ROW)
//                    .build();

            //在表中设置列簇
            test2.setColumnFamily(info);

            //使用admin来创建表
            admin.createTable(test2.build());

            //输出提示语句，并输出表名
            /**
             * 注意：所有的数据信息在hbase中都被转成了hbase专有的字节数组类型，无法使用java的方式将其转字符串
             * hbase专门提供一个工具类去解析这样的特别的字节数组
             * Bytes
             */
//            System.out.println(Arrays.toString(test2.build().getTableName().getName()) + " 表创建成功！！");
            System.out.println(Bytes.toString(test2.build().getTableName().getName()) + " 表创建成功！！");

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 需求2：删除Hbase中的某一张表
     * 1、先禁用表
     * 2、再删除表
     */
    @Test
    public void deleteOneTable() {
        try {
            //先判断该表是否存在
            //tableExists(TableName tableName)
            boolean b = admin.tableExists(TableName.valueOf("test2"));

            if (b) {
                //先禁用表
                admin.disableTable(TableName.valueOf("test2"));

                //再删除表
                admin.deleteTable(TableName.valueOf("test2"));

                System.out.println(Bytes.toString(TableName.valueOf("test2").getName()) + " 表成功删除！！！");

            } else {
                System.out.println("该表不存在");
            }

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 需求3：向表中添加一列数据
     * <p>
     * put '表名','行键','列簇:列名','列值'
     * put 'test2','1001','info:name','刘天保'
     */
    @Test
    public void putOneData() {
        try {
            //1、获取表的对象
            TableName test2 = TableName.valueOf("test2");
            Table table = conn.getTable(test2);

            //将列的数据封装成Put对象
            //public Put(byte [] row) {
            Put put = new Put(Bytes.toBytes("1002"));

            //public Put addColumn(byte [] family, byte [] qualifier, byte [] value)
//            put.addColumn(Bytes.toBytes("info"),Bytes.toBytes("name"),Bytes.toBytes("刘天保"));
            //Put add(Cell cell)
            //Cell指的是一个单元格（行键，列簇，列名，列值）
            //KeyValue是Cell类实现子类
            //KeyValue(final byte [] row, final byte [] family,final byte [] qualifier, final byte [] value)
            KeyValue cell = new KeyValue(Bytes.toBytes("1002"), Bytes.toBytes("info"), Bytes.toBytes("name"), Bytes.toBytes("孔志伟"));
            put.add(cell);

            //void put(Put put)
            table.put(put);

            System.out.println(Bytes.toString(test2.getName()) + " 表一列数据添加完毕！！！");

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 需求4：从Hbase表中获取一条数据
     * <p>
     * get '表名','行键','列簇:列名'
     */
    @Test
    public void getOneData() {
        try {
            //将表名封装成TableName的对象
            TableName test2 = TableName.valueOf("test2");

            //获取表的对象
            Table table = conn.getTable(test2);

            //创建一个Get对象
            //Get(byte [] row) 传入的是行键，因为行键是每一行的唯一标识，可以通过行键定位到某一行
            Get get = new Get(Bytes.toBytes("1001"));

            //表对象还需要通过方法获取到对应get结果
            Result result = table.get(get);

            //public byte[] getValue(byte [] family, byte [] qualifier)
//            String res = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("name")));
//            System.out.println("获取到一列数据：" + res);

            //获取所有的单元格，表示获取一行数据中的所有列,将每一列封装成Cell对象添加到List集合中
            List<Cell> cells = result.listCells();
            for (Cell cell : cells) {
                //CellUtil是Hbase提供的一个工具类，可以获取Cell单元格中四个维度（行键，列簇，列名，列值）
                String rowKey = Bytes.toString(CellUtil.cloneRow(cell));
                String family = Bytes.toString(CellUtil.cloneFamily(cell));
                String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
                String value = Bytes.toString(CellUtil.cloneValue(cell));
                System.out.println("行键：" + rowKey + ", 列簇：" + family + ", 列名：" + qualifier + ", 列值：" + value);
            }


        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 需求5：将学生信息表
     * 1500100001,施笑槐,22,女,文科六班
     * <p>
     * 考虑：行键怎么设计？
     * 唯一性
     * 长度不宜过长
     * 散列性
     * <p>
     * 对于此刻，将学号作为rowkey
     */
    @Test
    public void putAllData() {
        try {
            //额外的考虑：如果表中有几条脏数据，我们可以先删除脏数据
            //创建Delete对象
//            Delete d1 = new Delete(Bytes.toBytes("1001"));
            //...
            //先将表名封装成TableName对象
            TableName students = TableName.valueOf("students");
            //获取到表的对象
            Table table = conn.getTable(students);

//            table.delete(d1);
//            System.out.println("成功删除一行数据");
            //提前创建一个List集合对象，封装多个Put对象，每一个put是一列数据
            ArrayList<Put> puts = new ArrayList<>();

            //读取文件
            BufferedReader br = new BufferedReader(new FileReader("data/students.txt"));

            //一次读取一行
            String line = null;
            while ((line = br.readLine()) != null) {
                //1500100001,施笑槐,22,女,文科六班
                //按照英文逗号切
                String[] infos = line.split(",");
                //将学号作为行键创建一个put对象
                Put put = new Put(Bytes.toBytes(infos[0]));
                put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("name"), Bytes.toBytes(infos[1]));
                put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("age"), Bytes.toBytes(infos[2]));
                put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("gender"), Bytes.toBytes(infos[3]));
                put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("clazz"), Bytes.toBytes(infos[4]));

                //将封装好的Put对象放到集合中
                puts.add(put);
            }

            table.put(puts);
            System.out.println("学生信息添加完毕！！！");


        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 需求6：查询上一个需求插入的学生信息
     * <p>
     * scan 'students'
     * <p>
     * 我们今后一张表中不可能就只存储几千条数据，实际上hbase中的表，会非常大，上亿条数据
     * 全部查询没有意义，实际查询过程中，会指定开始的位置和结束的位置（行键）
     */
    @Test
    public void scanData() {
        try {
            //先将表名封装成TableName的对象
            TableName students = TableName.valueOf("students");

            //获取表的实例
            Table table = conn.getTable(students);

            //创建一个Scan的实例对象
            Scan scan = new Scan(); //该对象创建出来的后，如果不做其他的设置，默认是从第一个数据一直查询到最后一个

            //老版本写法
//            scan.setStartRow(Bytes.toBytes("1500100001"));
            //新版本写法
            scan.withStartRow(Bytes.toBytes("1500100011"));

//            scan.setStopRow(Bytes.toBytes("1500100011"));
            scan.withStopRow(Bytes.toBytes("1500100021"));

            //ResultScanner getScanner(Scan scan)
            ResultScanner resultScanner = table.getScanner(scan);

            //定义一个变量是Result类型的
            Result rs = null;
            //将每一次获取的值赋值给rs,如果rs不为null,说明获取到值
            //rs获取的是一行的值，而一行可以有若干个列
            while ((rs = resultScanner.next()) != null) {
                //获取到行键值
                String id = Bytes.toString(rs.getRow());
//                //获取每一列的值
//                String name = Bytes.toString(rs.getValue(Bytes.toBytes("info"), Bytes.toBytes("姓名")));
//                String age = Bytes.toString(rs.getValue(Bytes.toBytes("info"), Bytes.toBytes("年龄")));
//                String gender = Bytes.toString(rs.getValue(Bytes.toBytes("info"), Bytes.toBytes("性别")));
//                String clazz = Bytes.toString(rs.getValue(Bytes.toBytes("info"), Bytes.toBytes("班级")));
//
//                System.out.println("学号：" + id + ", 姓名：" + name + ", 年龄：" + age + ", 性别：" + gender + ", 班级：" + clazz);

                //上面的获取，是在我们已经知道了列名的前提下获取的，如果我们不知道列名，又怎么获取呢？
                //获取一行中所有的列，每一列组成了Cell对象，返回一个集合
                List<Cell> cells = rs.listCells();

                StringBuilder sb = new StringBuilder();
                sb.append("学号：").append(id);

                for (Cell cell : cells) {
                    String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
                    String value = Bytes.toString(CellUtil.cloneValue(cell));
                    sb.append(", ").append(qualifier).append("：").append(value);
                }

                //得到一行的拼接结果
                System.out.println(sb);
            }


        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 需求7：预分region
     * <p>
     * 在学习hbase的时候，我们向hbase中写入数据的时候，一开始只会向同一个region中插入数据，如果这个时候
     * 有大量的数据同时插入，就会导致该regionserver的压力过大，这样的问题叫做热点问题，解决这个热点问题的方案：预分region
     * 提前在建表的时候，划分分割点，预分多个region被多个regionserver所管理。
     * 方法：在建表的时候，提供分割点
     */
    @Test
    public void preRegionTable() {
        try {

            //pre_table
            //将表名封装成TableName的对象
            TableName preTable = TableName.valueOf("pre_table");

            //先判断表是否存在
            boolean b = admin.tableExists(preTable);
            if (b) {
                //表已经存在了，删除表
                admin.disableTable(preTable); //禁用表
                admin.deleteTable(preTable); //删除表
            }

            //如果不存在就直接创建
            //创建一个表描述器
            TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(preTable);
            //创建列簇描述器
//            ColumnFamilyDescriptor info = ColumnFamilyDescriptorBuilder.of("info");
            //给列簇设置过期时间,版本号，然后构建
            ColumnFamilyDescriptor info = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes("info"))
                    .setTimeToLive(60 * 60 * 24)
                    .setMaxVersions(3)
                    .build();

            //将列簇添加到表中
            tableDescriptorBuilder.setColumnFamily(info);

            //创建一个二维数组，表示分割点
            byte[][] splits = {
                    Bytes.toBytes("g"),
                    Bytes.toBytes("m"),
                    Bytes.toBytes("w"),
            };

            //创建表的时候，传入分割点
            admin.createTable(tableDescriptorBuilder.build(),splits);

            System.out.println("预分region表创建成功！！！");



        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    @After
    public void close() {
        if (admin != null) {
            try {
                admin.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        if (conn != null) {
            try {
                conn.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

}
